package cn.ultrarisk.scadmin.manager.biz.zookeeper.dao;

import cn.ultrarisk.mybatis.extension.query.Criteria;
import cn.ultrarisk.mybatis.extension.query.Query;
import cn.ultrarisk.scadmin.manager.biz.zookeeper.entity.ZkNode;
import cn.ultrarisk.scadmin.manager.biz.zookeeper.operator.ZookeeperOperator;
import cn.ultrarisk.scadmin.manager.support.BaseDAO;
import cn.ultrarisk.scadmin.manager.util.CollectionUtil;
import cn.ultrarisk.scadmin.manager.util.JsonUtil;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Repository;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * ZK节点数据操作类
 *
 * @author sevendlong(lin23871@163.com)
 * @date 2014/3/28
 * @since V1.0
 */
@Repository
public class ZkNodeDAO extends BaseDAO<ZkNode, Integer> {
    @Autowired
    private ZookeeperOperator zookeeperOperator;

    /**
     * 构造器
     */
    public ZkNodeDAO() {
        super(ZkNode.class, ZkNode.COLUMN_PROPERTY_MAP);
    }

    private List<ZkNode> assembly(List<ZkNode> nodes) {
        Map<Integer, ZkNode> nodeMap = new HashMap<Integer, ZkNode>();
        for (ZkNode node : nodes) {
            nodeMap.put(node.getId(), node);
        }
        for (ZkNode node : nodes) {
            ZkNode parent = nodeMap.get(node.getParentId());
            if (parent != null) {
                parent.addChild(node);
            }
        }

        List<ZkNode> newNodes = new ArrayList<ZkNode>();
        for (ZkNode node : nodes) {
            if (nodeMap.get(node.getParentId()) == null) {
                newNodes.add(node);
            }
        }
        return newNodes;
    }

    public ZkNode getFromZk(String path, ZkNode parent) {
        Stat stat = zookeeperOperator.getStat(path);
        ZkNode node = new ZkNode(stat);
        node.setPath(path);
        node.setData(zookeeperOperator.getData(node.getPath(), stat));
        node.setParent(parent);
        if (zookeeperOperator.hasChild(stat)) {
            List<String> childPaths = zookeeperOperator.getChildren(path);
            if (CollectionUtil.isNotEmpty(childPaths)) {
                for (String childPath : childPaths) {
                    node.addChild(getFromZk(childPath, node));
                }
            }
        }
        return node;
    }

    public ZkNode getByPath(String path, Integer backupId) {
        return get(Query.query("path", path).eq(ZkNode.BACKUP_ID, backupId));
    }

    public List<ZkNode> findChildrenByPath(String path, Integer backupId) {
        ZkNode parent = getByPath(path, backupId);
        return parent == null ? null : findList(Query.query(ZkNode.PARENT_ID, parent.getId()));
    }

    public List<ZkNode> findByLikePath(String likePath, Integer backupId, boolean assembly, String... tableColumns) {
        Query query = Query.query(ZkNode.BACKUP_ID, backupId).like(ZkNode.PATH, likePath);
        query.addSort(ZkNode.PATH, Query.Direction.ASC);
        List<ZkNode> nodes = findList(query, tableColumns);
        return assembly ? assembly(nodes) : nodes;
    }

    public List<ZkNode> findByLikePaths(List<String> likePaths, Integer backupId, boolean assembly, String... tableColumns) {
        Query query = Query.query(ZkNode.BACKUP_ID, backupId);

        List<Criteria> subCriterias = new ArrayList<Criteria>();
        for (String likePath : likePaths) {
            subCriterias.add(new Criteria(ZkNode.PATH, Criteria.Opt.LIKE, likePath));
        }
        query.addSubCriteria(Query.Logical.OR, subCriterias);
        query.addSort(ZkNode.PATH, Query.Direction.ASC);
        List<ZkNode> nodes = findList(query, tableColumns);
        return assembly ? assembly(nodes) : nodes;
    }

    public List<ZkNode> findByBackupId(Integer backupId, boolean assembly, String... tableColumns) {
        List<ZkNode> nodes = findList(Query.query(ZkNode.BACKUP_ID, backupId), tableColumns);
        return assembly ? assembly(nodes) : nodes;
    }

    public Map<String, String> getCollectionUniqueKeyMap(Integer backupId) {
        Map<String, ZkNode> collectionNodeMap = new HashMap<String, ZkNode>();
        List<ZkNode> collectionNodes = findChildrenByPath("/collections", backupId);
        if (collectionNodes != null) {
            for (ZkNode collectionNode : collectionNodes) {
                collectionNodeMap.put(collectionNode.getRelativePath(), collectionNode);
            }
        }
        Map<String, ZkNode> schemaNodeMap = new HashMap<String, ZkNode>();
        List<ZkNode> schemaNodes = findByLikePath("/configs/%/schema.xml", backupId, true);
        if (schemaNodes != null) {
            for (ZkNode schemaNode : schemaNodes) {
                String configName = schemaNode.getPath().replace("/configs/", "").replace("/schema.xml", "");
                schemaNodeMap.put(configName, schemaNode);
            }
        }
        Map<String, String> uniqueKeyMap = new HashMap<String, String>();
        Pattern pattern = Pattern.compile("<uniqueKey>([\\S\\s]*)</uniqueKey>");
        for (String collectionName : collectionNodeMap.keySet()) {
            ZkNode collectionNode = collectionNodeMap.get(collectionName);
            if (collectionNode == null) {
                continue;
            }
            String configName = JsonUtil.fromJsonAsMap(collectionNode.getData()).get("configName").toString();
            if (configName == null) {
                continue;
            }
            ZkNode schemaNode = schemaNodeMap.get(configName);
            if (schemaNode != null && schemaNode.getDataLength() > 0) {
                Matcher matcher = pattern.matcher(schemaNode.getData());
                if (matcher.find()) {
                    uniqueKeyMap.put(collectionName, matcher.group(1).trim());
                }
            }
        }
        return uniqueKeyMap;
    }
}
